package com.henry.config;

import jakarta.annotation.Resource;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.MQAdmin;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.autoconfigure.*;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.condition.*;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.core.env.Environment;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

/**
 * @author Liu Hanlin
 * @create 2025-11-05 22:36
 */
@Configuration
@EnableConfigurationProperties({CustomerRocketMQProperties.class})
@ConditionalOnClass({MQAdmin.class})
@ConditionalOnProperty(
        prefix = "rocketmq",
        value = {"name-server"},
        matchIfMissing = true
)
@Import(
        {
            CustomerMessageConverterConfiguration.class, ListenerContainerConfiguration.class,
            ExtProducerResetConfiguration.class, ExtConsumerResetConfiguration.class,
            RocketMQTransactionConfiguration.class, RocketMQListenerConfiguration.class
        }
)
@AutoConfigureAfter({CustomerMessageConverterConfiguration.class})
@AutoConfigureBefore({RocketMQTransactionConfiguration.class, RocketMQAutoConfiguration.class})
public class CustomerRocketMQAutoConfiguration implements ApplicationContextAware {
    private static final Logger log = LoggerFactory.getLogger(CustomerRocketMQAutoConfiguration.class);
    public static final String ROCKETMQ_TEMPLATE_DEFAULT_GLOBAL_NAME = "rocketMQTemplate";
    public static final String PRODUCER_BEAN_NAME = "defaultMQProducer";
    public static final String CONSUMER_BEAN_NAME = "defaultLitePullConsumer";
    private final Environment environment;

    @Resource
    private CustomerRocketMQProperties customerRocketMQProperties;
    private ApplicationContext applicationContext;

    public CustomerRocketMQAutoConfiguration(Environment environment) {
        this.environment = environment;
        this.checkProperties();
    }

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    public void checkProperties() {
        String nameServer = (String)this.environment.getProperty("rocketmq.name-server", String.class);
        log.debug("rocketmq.nameServer = {}", nameServer);
        if (nameServer == null) {
            log.warn("The necessary spring property 'rocketmq.name-server' is not defined, all rockertmq beans creation are skipped!");
        }

    }

    @Bean({PRODUCER_BEAN_NAME})
    @ConditionalOnMissingBean({DefaultMQProducer.class})
    @ConditionalOnProperty(
            prefix = "rocketmq",
            value = {"name-server", "producer.group"}
    )
    public DefaultMQProducer defaultMQProducer() {
        CustomerRocketMQProperties.Producer producerConfig = customerRocketMQProperties.getProducer();
        String nameServer = customerRocketMQProperties.getNameServer();
        String groupName = producerConfig.getGroup();
        Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
        Assert.hasText(groupName, "[rocketmq.producer.group] must not be null");
        String accessChannel = customerRocketMQProperties.getAccessChannel();
        String ak = customerRocketMQProperties.getProducer().getAccessKey();
        String sk = customerRocketMQProperties.getProducer().getSecretKey();
        boolean isEnableMsgTrace = customerRocketMQProperties.getProducer().isEnableMsgTrace();
        String customizedTraceTopic = customerRocketMQProperties.getProducer().getCustomizedTraceTopic();
        DefaultMQProducer producer = RocketMQUtil.createDefaultMQProducer(groupName, ak, sk, isEnableMsgTrace, customizedTraceTopic);
        producer.setNamesrvAddr(nameServer);
        if (StringUtils.hasLength(accessChannel)) {
            producer.setAccessChannel(AccessChannel.valueOf(accessChannel));
        }

        producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
        producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
        producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
        producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
        producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
        producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
        producer.setUseTLS(producerConfig.isTlsEnable());
        if (StringUtils.hasText(producerConfig.getNamespace())) {
            producer.setNamespace(producerConfig.getNamespace());
        }

//        if (StringUtils.hasText(producerConfig.getNamespaceV2())) {
//            producer.setNamespaceV2(producerConfig.getNamespaceV2());
//        }

        producer.setInstanceName(producerConfig.getInstanceName());
        log.info("a producer ({}) init on namesrv {}", groupName, nameServer);
        return producer;
    }

    @Bean({CONSUMER_BEAN_NAME})
    @ConditionalOnMissingBean({DefaultLitePullConsumer.class})
    @ConditionalOnProperty(
            prefix = "rocketmq",
            value = {"name-server", "pull-consumer.group", "pull-consumer.topic"}
    )
    public DefaultLitePullConsumer defaultLitePullConsumer() throws MQClientException {
        CustomerRocketMQProperties.PullConsumer consumerConfig = customerRocketMQProperties.getPullConsumer();
        String nameServer = customerRocketMQProperties.getNameServer();
        String groupName = consumerConfig.getGroup();
        String topicName = consumerConfig.getTopic();
        Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
        Assert.hasText(groupName, "[rocketmq.pull-consumer.group] must not be null");
        Assert.hasText(topicName, "[rocketmq.pull-consumer.topic] must not be null");
        String accessChannel = customerRocketMQProperties.getAccessChannel();
        MessageModel messageModel = MessageModel.valueOf(consumerConfig.getMessageModel());
        SelectorType selectorType = SelectorType.valueOf(consumerConfig.getSelectorType());
        String selectorExpression = consumerConfig.getSelectorExpression();
        String ak = consumerConfig.getAccessKey();
        String sk = consumerConfig.getSecretKey();
        int pullBatchSize = consumerConfig.getPullBatchSize();
        boolean useTLS = consumerConfig.isTlsEnable();
        DefaultLitePullConsumer litePullConsumer = RocketMQUtil.createDefaultLitePullConsumer(nameServer, accessChannel, groupName, topicName, messageModel, selectorType, selectorExpression, ak, sk, pullBatchSize, useTLS);
        litePullConsumer.setEnableMsgTrace(consumerConfig.isEnableMsgTrace());
        litePullConsumer.setCustomizedTraceTopic(consumerConfig.getCustomizedTraceTopic());
        if (StringUtils.hasText(consumerConfig.getNamespace())) {
            litePullConsumer.setNamespace(consumerConfig.getNamespace());
        }

//        if (StringUtils.hasText(consumerConfig.getNamespaceV2())) {
//            litePullConsumer.setNamespaceV2(consumerConfig.getNamespaceV2());
//        }

        litePullConsumer.setInstanceName(consumerConfig.getInstanceName());
        log.info("a pull consumer({} sub {}) init on namesrv {}", new Object[]{groupName, topicName, nameServer});
        return litePullConsumer;
    }

    @Bean(
            destroyMethod = "destroy"
    )
    @Conditional({ProducerOrConsumerPropertyCondition.class})
    @ConditionalOnMissingBean(
            name = {ROCKETMQ_TEMPLATE_DEFAULT_GLOBAL_NAME}
    )
    public RocketMQTemplate rocketMQTemplate(RocketMQMessageConverter rocketMQMessageConverter) {
        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
        if (this.applicationContext.containsBean("defaultMQProducer")) {
            rocketMQTemplate.setProducer((DefaultMQProducer)this.applicationContext.getBean("defaultMQProducer"));
        }

        if (this.applicationContext.containsBean("defaultLitePullConsumer")) {
            rocketMQTemplate.setConsumer((DefaultLitePullConsumer)this.applicationContext.getBean("defaultLitePullConsumer"));
        }

        rocketMQTemplate.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
        return rocketMQTemplate;
    }

    static class ProducerOrConsumerPropertyCondition extends AnyNestedCondition {
        public ProducerOrConsumerPropertyCondition() {
            super(ConfigurationPhase.REGISTER_BEAN);
        }

        @ConditionalOnBean({DefaultLitePullConsumer.class})
        static class DefaultLitePullConsumerExistsCondition {
            DefaultLitePullConsumerExistsCondition() {
            }
        }

        @ConditionalOnBean({DefaultMQProducer.class})
        static class DefaultMQProducerExistsCondition {
            DefaultMQProducerExistsCondition() {
            }
        }
    }
}

